package org.example.myframework.flink.demo;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.apache.flink.streaming.api.TimeCharacteristic;

public class WordCountDemo {
    public static void main(String[] args) throws Exception {
        // 创建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置执行模式为远程模式
        env.setParallelism(1); // 设置并行度
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); // 设置时间特性，可选

        // 设置数据源，这里使用从socket读取的数据
        DataStream<String> text = env.socketTextStream("localhost", 9999);

        // 对输入流进行处理
        DataStream<Tuple2<String, Integer>> counts = text
                // 拆分每一行并转换成 (word, 1) 的元组
                .flatMap(new Tokenizer())
                // 按照单词分组并进行求和操作
                .keyBy(0)
                .sum(1);

        // 输出结果
        counts.print();

        // 启动作业
        env.execute("WordCount Demo");
    }

    // 自定义 FlatMapFunction 实现拆分单词并生成 (word, 1) 元组
    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            // 将每一行文本按空格拆分为单词
            String[] words = value.toLowerCase().split("\\W+");

            // 发出每个单词的计数为 1 的元组
            for (String word : words) {
                if (word.length() > 0) {
                    out.collect(new Tuple2<>(word, 1));
                }
            }
        }
    }
}
